package com.kk.demo.java8.threadstream;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.LinkedBlockingQueue;

public class Use implements Runnable {

    private LinkedBlockingQueue<String> value;

    public Use(LinkedBlockingQueue<String> value) {
        this.value = value;
    }

    @Override
    public void run() {
        
        while(true)
        {
            Map<Integer, Integer> countrsp = new HashMap<>();
            value.stream().map(this::getMapPrcess).reduce(countrsp,this::reduceFun);
            
            System.out.println(countrsp);
        }

    }

    /**
     * 得到进行map处理的方法
     * @param key 输入的行记录
     * @return 结果的map信息
     */
    public Map<Integer, Integer> getMapPrcess(String key) {
        String[] arrayks = key.split(",");
        int a1 = Integer.parseInt(arrayks[0]);
        int a2 = Integer.parseInt(arrayks[1]);

        Map<Integer, Integer> map = new HashMap<>();

        map.put(a1, a2);

        return map;
    }

    /**
     * 进行reduce计算的方法
     * @param a 输入参数信息
     * @param b 当前的map信息
     * @return 结果信息
     */
    public Map<Integer, Integer> reduceFun(Map<Integer, Integer> a,
            Map<Integer, Integer> b) {

        if (null != a && null != b) {
            Iterator<Entry<Integer, Integer>> iterReduce = b.entrySet()
                    .iterator();

            Entry<Integer, Integer> entry = null;

            while (iterReduce.hasNext()) {
                entry = iterReduce.next();

                // 检查是否包含b中的key信息
                if (a.containsKey(entry.getKey())) {
                    int value = a.get(entry.getKey());

                    a.put(entry.getKey(), entry.getValue() + value);
                }
                // 如果不包含，则新加入
                else {
                    a.put(entry.getKey(), entry.getValue());
                }
            }
        }

        return a;
    }

}
